package com.ikas.ai.server.kafka;

import cn.hutool.cache.CacheUtil;
import cn.hutool.cache.impl.TimedCache;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.JSONValidator;
import com.ikas.ai.enums.MeteTypeEnum;
import com.ikas.ai.enums.redis.RedisKeyEnum;
import com.ikas.ai.model.KafkaMeteData;
import com.ikas.ai.server.module.steady.enums.SteadyMeteTypeEnum;
import com.ikas.ai.server.module.steady.service.SteadyMonitorMeteService;
import com.ikas.ai.utils.RedisUtils;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.*;

/**
 * 稳态数据消费消息
 *
 * @author hubowei
 */
@Slf4j
@Component
public class KafkaSteadyMeteConsumerService {

    /**
     * 本地缓存时间 （位：毫秒）
     */
    @Value("${local-cache-timeout:3600000}")
    private Long localCacheTimeout;

    /**
     * 本地缓存
     */
    private TimedCache<String, Map<String, Integer>> localTimeCache;


    @Resource
    private SteadyMonitorMeteService steadyMonitorMeteService;

    /**
     * 获取本地缓存对象
     *
     * @return Boolean
     */
    public TimedCache<String, Map<String, Integer>> getLocalTimeCache() {
        if (localTimeCache == null) {
            localTimeCache = CacheUtil.newTimedCache(localCacheTimeout);
        }
        return localTimeCache;
    }

    /**
     * 消费UData数据到稳态中
     *
     * @param list 104数据
     * @param ack  应答参数
     */
    @KafkaListener(topics = {KafkaInitialConfiguration.TOPIC_U_DATA}, topicPattern = "0", concurrency = "1",
            groupId = KafkaInitialConfiguration.TOPIC_GROUP_STEADY,
            properties = {
                    "enable.auto.commit:false", "max.poll.interval.ms:60000"})
    @Transactional(rollbackFor = Exception.class)
    public void consumerUdataToSteady(List<ConsumerRecord<?, ?>> list, Acknowledgment ack) {
        try {
            handleDataToSteady(list);
        } catch (Exception e) {
            log.error("message:{}", e.getMessage());
        } finally {
            ack.acknowledge();
        }
    }

    /**
     * 将kafka数据转换从实体对象
     *
     * @param list
     * @return
     */
    private List<KafkaMeteData> coverRecordToMeteDataEntity(List<ConsumerRecord<?, ?>> list) {
        List<KafkaMeteData> meteDataList = new ArrayList<>();
        for (ConsumerRecord<?, ?> record : list) {
            final Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) {
                final String message = (String) kafkaMessage.get();
                // 兼容kafka数据两种json
                final JSONValidator.Type type = JSONValidator.from(message).getType();
                if (type == JSONValidator.Type.Array) {
                    meteDataList.addAll(JSONArray.parseArray(message, KafkaMeteData.class));
                } else if (type == JSONValidator.Type.Object) {
                    meteDataList.add(JSONObject.parseObject(message, KafkaMeteData.class));
                } else {
                    continue;
                }
            }
        }
        return meteDataList;
    }

    @SneakyThrows
    private boolean handleDataToSteady(List<ConsumerRecord<?, ?>> list) {
        //1.将kafka数据转换从实体对象
        List<KafkaMeteData> kafkaMeteDataList = coverRecordToMeteDataEntity(list);
        //2.将kafka数据进行稳态相关处理
        syncMeteDataToSteady(kafkaMeteDataList);
        return true;
    }

    /**
     * 将kafka数据进行稳态相关处理
     *
     * @param kafkaMeteDataList kafka data
     */
    private void syncMeteDataToSteady(List<KafkaMeteData> kafkaMeteDataList) {
        for (KafkaMeteData meteData : kafkaMeteDataList) {
            //从redis获取值判断是稳态测点还是监测测点
            Integer steadyMeteTypeVal = checkInSteadyMeteType(meteData);
            if (Objects.isNull(steadyMeteTypeVal)) {
                continue;
            }
            //1.如果是遥信且是稳态测点
            if (MeteTypeEnum.YX.getNum().equals(meteData.getMeteType()) &&
                    SteadyMeteTypeEnum.STATE_METE.getValue().equals(steadyMeteTypeVal)) {
                log.info("稳态测点:{}", JSON.toJSONString(meteData));
                steadyMonitorMeteService.handleKafkaDataOfStateMeteType(meteData);
            } else if (MeteTypeEnum.YC.getNum().equals(meteData.getMeteType()) &&
                    SteadyMeteTypeEnum.MONITOR_METE.getValue().equals(steadyMeteTypeVal)) {
                log.info("监测测点:{}", JSON.toJSONString(meteData));
                //2.如果是遥测且是监测测点
                steadyMonitorMeteService.handleKafkaDataOfMonitorMeteType(meteData);
            }
        }
    }

    private Integer checkInSteadyMeteType(KafkaMeteData meteData) {
        //校验和初始化本地缓存
        checkAndInitLocalTimeCache();
        Map<String, Integer> map = getLocalTimeCache().get(RedisKeyEnum.STEADY_METE_CODE.key());
        if (ObjectUtils.isNotEmpty(map) && map.containsKey(meteData.getMeteCode())) {
            return map.get(meteData.getMeteCode());
        }
        return null;
    }

    /**
     * 校验和初始化本地缓存
     */
    private void checkAndInitLocalTimeCache() {
        TimedCache<String, Map<String, Integer>> localTimeCache = getLocalTimeCache();
        Map<String, Integer> map = localTimeCache.get(RedisKeyEnum.STEADY_METE_CODE.key());
        if (ObjectUtils.isEmpty(map)) {
            //初始化本地缓存
            initLocalTimeCache(localTimeCache);
        }
    }

    /**
     * 初始化本地缓存
     *
     * @param localTimeCache 本地缓存对象
     */
    private void initLocalTimeCache(TimedCache<String, Map<String, Integer>> localTimeCache) {
        Map<String, Integer> redisMap = ObjectUtils.defaultIfNull(
                RedisUtils.get(RedisKeyEnum.STEADY_METE_CODE.key()), new HashMap<>()
        );
        log.info("初始化稳态测点本地缓存，来源redis，数据size：{}", redisMap.size());
        localTimeCache.put(RedisKeyEnum.STEADY_METE_CODE.key(), redisMap);
    }
}



















